package com.sunl.mall.ali.mq.producer;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
import com.aliyun.openservices.ons.api.transaction.TransactionProducer;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
import com.sunl.mall.ali.mq.api.RocketMqLocalTransactionMessageChecker;
import com.sunl.mall.ali.mq.api.producer.RocketMqLocalTransactionExecutor;
import com.sunl.mall.ali.mq.api.producer.RocketMqTransactionProducer;
import com.sunl.mall.ali.mq.model.RocketMqMessageBody;
import com.sunl.mall.ali.mq.model.RocketMqMessageContext;
import com.sunl.mall.ali.mq.model.transaction.RocketMqTransactionMessageContext;
import com.sunl.mall.ali.mq.model.transaction.RocketMqTransactionMessageWrapper;
import com.sunl.mall.ali.mq.model.transaction.RocketMqTransactionSendResult;
import com.sunl.mall.ali.mq.serializer.RocketMqSerializer;
import com.sunl.mall.ali.mq.util.MqContextUtil;
import com.sunl.mall.common.api.util.ExceptionUtil;
import com.sunl.mall.common.api.util.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class RocketMqTransactionProducerBean extends AbstractRocketMqProducer implements RocketMqTransactionProducer {
    private final static Logger log = LoggerFactory.getLogger(RocketMqTransactionProducerBean.class);

    /**
     * 需要注入该字段，指定构造{@link TransactionProducer}实例的属性，具体支持的属性详见{@link PropertyKeyConst}
     */
    private final Properties properties;

    private RocketMqLocalTransactionMessageChecker rocketMqLocalTransactionMessageChecker;

    private TransactionProducer transactionProducer;

    public RocketMqTransactionProducerBean(Properties properties) {
        this.properties = properties;
    }

    @Override
    public <T> boolean send(RocketMqTransactionMessageWrapper<T> message, RocketMqLocalTransactionExecutor<T> executor, RocketMqSerializer<T> serializer) {
        return sendBackResult(message, executor, serializer).isSuccess();
    }

    @Override
    public <T> boolean send(RocketMqTransactionMessageWrapper<T> message, RocketMqLocalTransactionExecutor<T> executor, Object args, RocketMqSerializer<T> serializer) {
        return sendBackResult(message, executor, args, serializer).isSuccess();
    }

    @Override
    public <T> RocketMqTransactionSendResult sendBackResult(RocketMqTransactionMessageWrapper<T> message, RocketMqLocalTransactionExecutor<T> executor, RocketMqSerializer<T> serializer) {
        return sendBackResult(message, executor, new Object(), serializer);
    }

    @Override
    public <T> RocketMqTransactionSendResult sendBackResult(RocketMqTransactionMessageWrapper<T> message, RocketMqLocalTransactionExecutor<T> executor, Object args, RocketMqSerializer<T> serializer) {
        if (message.getContent() == null) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey()};
            log.error("Transaction Mq send Failure, Because Message content is null , topic:[{}], tag:[{}], key:[{}]", params);
            return RocketMqTransactionSendResult.fail();
        }
        message.putUserProperties(PropertyKeyConst.CheckImmunityTimeInSeconds, Integer.toString(message.getCheckImmunityTimeInSeconds()));
        SendResult sendResult = null;
        final RocketMqTransactionSendResult transactionExceptionSendResult = new RocketMqTransactionSendResult();
        long start = SystemClock.currentTimeMillis();
        try {
            byte[] bytes = serializer.serialize(message.getContent());
            message.setBody(bytes);
            if (!checkSendMsgBody(message)) {
                return RocketMqTransactionSendResult.fail();
            }
            sendResult = this.transactionProducer.send(message, (msg, arg) -> {
                transactionExceptionSendResult.setSendHalfMsgOk(true);
                RocketMqTransactionMessageContext transactionMessageContext = MqContextUtil.getTransactionMessageContext(msg);
                TransactionStatus transactionStatus = TransactionStatus.Unknow;
                try {
                    transactionStatus = executor.execute(message, arg, transactionMessageContext);
                } catch (Throwable e) {
                    transactionExceptionSendResult.setLocalTransactionThrowable(e);
                    throw e;
                }
                return transactionStatus;
            }, args);
        } catch (Throwable e) {
            Object[] params = {message.getTopic(), message.getTag(), message.getKey(), message.getContent(), SystemClock.currentTimeMillis() - start, ExceptionUtil.getAsString(e)};
            log.error("Transaction Mq send Exception, topic:[{}], tag:[{}], key:[{}], message:[{}], costTime:{}ms, Some Exception Occur:[{}]", params);
        }
        RocketMqTransactionSendResult result = RocketMqTransactionSendResult.successIfNotNull(sendResult);
        result.setLocalTransactionThrowable(transactionExceptionSendResult.getLocalTransactionThrowable());
        result.setSendHalfMsgOk(transactionExceptionSendResult.isSendHalfMsgOk());
        return result;
    }

    @Override
    public boolean isStarted() {
        return this.transactionProducer.isStarted();
    }

    @Override
    public boolean isClosed() {
        return this.transactionProducer.isClosed();
    }

    @Override
    public void start() {
        if (null == this.properties) {
            throw new ONSClientException("properties not set");
        }
        if (null == this.rocketMqLocalTransactionMessageChecker) {
            throw new ONSClientException("Transaction MQ localTransactionChecker must not null");
        }
        init(this.properties);
        this.transactionProducer = ONSFactory.createTransactionProducer(this.properties, wrapperLocalTransactionChecker(rocketMqLocalTransactionMessageChecker));
        this.transactionProducer.start();
    }

    @Override
    public void updateCredential(Properties credentialProperties) {
        if (this.transactionProducer != null) {
            this.transactionProducer.updateCredential(credentialProperties);
        }
    }

    @Override
    public void shutdown() {
        if (this.transactionProducer != null) {
            this.transactionProducer.shutdown();
        }
    }

    /**
     * 转换自定义事务回查逻辑到sdk的事务回查逻辑
     *
     * @param checker
     * @param <T>
     * @return
     */
    private <T> LocalTransactionChecker wrapperLocalTransactionChecker(final RocketMqLocalTransactionMessageChecker<T> checker) {
        return message -> {
            RocketMqMessageBody<T> record = null;
            try {
                RocketMqMessageContext messageContext = MqContextUtil.getMessageContext(message);
                record = checker.deserialize(message.getBody(), messageContext);
                if (record == null) {
                    Object[] params = {message.getMsgID(), message.getTopic(), message.getTag(), message.getKey(), RocketMqTransactionProducerBean.this.properties.get("GROUP_ID"), message.getReconsumeTimes()};
                    log.error("Transaction Mq consume message Exception: record is null. msgId:[{}], topic[{}], tag[{}], key[{}], consumerId:[{}], retryTimes:[{}]", params);
                } else if (RocketMqTransactionProducerBean.log.isDebugEnabled()) {
                    Object[] params = {message.getMsgID(), message.getTopic(), message.getTag(), message.getKey(), RocketMqTransactionProducerBean.this.properties.get("GROUP_ID"), message.getReconsumeTimes(), record};
                    log.debug("Transaction Mq Received message: msgId:[{}], topic[{}], tag[{}], key[{}], consumerId:[{}], retryTimes:[{}], record:[{}]", params);
                } else {
                    Object[] params = {message.getMsgID(), message.getTopic(), message.getTag(), message.getKey(), message.getReconsumeTimes()};
                    log.info("Transaction Mq Received message: msgId:[{}], topic[{}], tag[{}], key[{}], retryTimes:[{}]", params);
                }
                return checker.check(record, messageContext);
            } catch (Throwable e) {
                if (record != null) {
                    Object[] params = {message.getMsgID(), message.getTopic(), message.getTag(), message.getKey(), RocketMqTransactionProducerBean.this.properties.get("GROUP_ID"), message.getReconsumeTimes(), record, ExceptionUtil.getAsString(e)};
                    RocketMqTransactionProducerBean.log.error("Transaction Mq consume message Exception: msgId:[{}], topic[{}], tag[{}], key[{}], consumerId:[{}], retryTimes:[{}], record:[{}], Exception:{}", params);
                } else {
                    Object[] params = {message.getMsgID(), message.getTopic(), message.getTag(), message.getKey(), RocketMqTransactionProducerBean.this.properties.get("GROUP_ID"), message.getReconsumeTimes(), null, ExceptionUtil.getAsString(e)};
                    RocketMqTransactionProducerBean.log.error("Transaction Mq consume message Exception: msgId:[{}], topic[{}], tag[{}], key[{}], consumerId:[{}], retryTimes:[{}], record is null, message:[{}], Exception:{}", params);
                }
                return TransactionStatus.Unknow;
            }
        };
    }

    public RocketMqLocalTransactionMessageChecker getRocketMqTransactionMessageChecker() {
        return rocketMqLocalTransactionMessageChecker;
    }

    public void setRocketMqTransactionMessageChecker(RocketMqLocalTransactionMessageChecker rocketMqLocalTransactionMessageChecker) {
        this.rocketMqLocalTransactionMessageChecker = rocketMqLocalTransactionMessageChecker;
    }
}
